package com.bluemoon.model.test.validation.service.impl;

import cn.com.bluemoon.daps.common.domain.ResultBean;
import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.BmStatus;
import cn.com.bluemoon.daps.common.enums.DatabaseType;
import cn.com.bluemoon.daps.common.exception.DapException;
import cn.com.bluemoon.daps.model.dto.DapDataModelTableDto;
import cn.com.bluemoon.daps.model.entity.*;
import cn.com.bluemoon.daps.model.mapper.DapDataModelRecordMapper;
import cn.com.bluemoon.daps.model.service.*;
import cn.com.bluemoon.daps.model.utils.contrast.VersionContrast;
import cn.com.bluemoon.daps.model.vo.SearchModelApplicationVo;
import cn.com.bluemoon.daps.standard.dto.ApplicationModelDto;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bluemoon.model.test.validation.service.DapDataModelRecordService;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class DapDataModelRecordServiceImpl extends ServiceImpl<DapDataModelRecordMapper, DapDataModelRecord> implements DapDataModelRecordService {

    @Resource
    DapDataModelTableService dapDataModelTableService;

    @Resource
    DapDataModelTableFieldService dapDataModelTableFieldService;

    @Resource
    DapDataModelSubordinateKeyService dapDataModelSubordinateKeyService;

    @Resource
    DapDataModelTablePartitionService dapDataModelTablePartitionService;

    @Resource
    DapDataModelOperateHistoryService dapDataModelOperateHistoryService;

    @Resource
    DapDataModelRecordMapper modelRecordMapper;

    @Resource
    DapDataModelVersionService dapDataModelVersionService;

    /**
     * v1.2
     * 发布列表
     * @param application
     * @return
     */
    @Override
    public IPage<ApplicationModelDto> getList(SearchModelApplicationVo application, Integer applicationType) {

        Page<ApplicationModelDto> pages = modelRecordMapper.selectApplicationList(new Page<>(application.getPageNum(), application.getPageSize()), application, applicationType);
        return pages;
    }

    /**
     * v1.2
     * 单个模型申请发布
     * @param recordId
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public boolean singleModelApplicationRelease(String recordId, String desc) {

        // 首先将状态改为待审批
        boolean update = this.update(new LambdaUpdateWrapper<DapDataModelRecord>()
                .set(DapDataModelRecord::getPublicStatus, 3)
                .set(DapDataModelRecord::getApplicationTime, LocalDateTime.now())
                .set(DapDataModelRecord::getApplicationBy, UserInfoHolder.getAccount())
                .set(DapDataModelRecord::getChangeDesc, desc)
                .eq(DapDataModelRecord::getId, recordId));
        return update;
    }

    /**
     * v1.2
     * 撤销
     * @param modelId
     * @param recordId
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public void cancelRelease(String modelId, String recordId) {
        // 首先删除此条记录
        modelRecordMapper.deleteById(recordId);
        // 首先判断是否为第一次进行发布
        List<DapDataModelRecord> dapDataModelRecords = getBaseMapper()
                .selectList(new LambdaQueryWrapper<DapDataModelRecord>()
                .eq(DapDataModelRecord::getModelId, modelId)
                .eq(DapDataModelRecord::getBmStatus, BmStatus.ON.getCode()));
        // 为空表示第一次进行发布
        if (dapDataModelRecords.isEmpty()) {
            // 删除复制的表
            dapDataModelTableService.deleteTable(modelId);
            // 删除备份的字段
            dapDataModelTableFieldService.deleteField(modelId);
            // 删除备份的索引
            dapDataModelSubordinateKeyService.deleteKey(modelId);
            // 删除备份的分区
            dapDataModelTablePartitionService.deletePartition(modelId);
        }
        // 删除历史表中的数据
        dapDataModelOperateHistoryService.getBaseMapper()
                .delete(new LambdaQueryWrapper<DapDataModelOperateHistory>()
                .eq(DapDataModelOperateHistory::getDataModelRecordId, recordId));
    }

    /**
     * v1.2
     * 不通过发布
     * @param recordId
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public void unPass(String recordId) {

        // 将状态设置成不通过
        this.update(new LambdaUpdateWrapper<DapDataModelRecord>()
                .set(DapDataModelRecord::getPublicStatus, 4)
                .set(DapDataModelRecord::getVerifyBy, UserInfoHolder.getAccount())
                .set(DapDataModelRecord::getReleaseTime, LocalDateTime.now())
                .eq(DapDataModelRecord::getId, recordId));

    }

    /**
     * v1.2
     * 通过发布
     * @param recordId
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public void pass(String recordId, String modelId, DatabaseType databaseType) {

        // 将状态置为已发布,更新发布时间
        modelRecordMapper.updatePublicTime(recordId, UserInfoHolder.getAccount());
        // 如果是第一次进行发布
        List<DapDataModelRecord> dapDataModelRecords = this.getBaseMapper()
                .selectList(new LambdaQueryWrapper<DapDataModelRecord>()
                .eq(DapDataModelRecord::getModelId, modelId)
                .eq(DapDataModelRecord::getBmStatus, BmStatus.ON.getCode()));
        // 最新发布时间
        LocalDateTime max = dapDataModelRecords.stream().filter(o -> o.getReleaseTime() != null)
                .map(DapDataModelRecord::getReleaseTime).distinct().max((e1, e2) -> e1.compareTo(e2)).get();
        // 存在发布版本
        if (dapDataModelRecords.size() > 1) {
            // 查询模型下属表, 新数据
            List<DapDataModelTableDto> tableList = dapDataModelTableService.getCopyTableInfo(modelId);
            List<DapDataModelTableDto> newList = getKeyAndPartitionList(tableList, databaseType, modelId, true);
            // 查询上版本的数据
            List<DapDataModelTableDto> tables = dapDataModelTableService.getCopyTableList(modelId, 2);
            List<DapDataModelTableDto> oldList = getKeyAndPartitionList(tables, databaseType, modelId, false);
            ResultBean<List<DapDataModelOperateHistory>> result = VersionContrast.modelContrast(newList, databaseType, oldList, max, true);
            Map<String, Object> map = result.getMoreData();
            // 新增的表
            List<DapDataModelTableDto> addTable = (List<DapDataModelTableDto>) map.get("addTable");
            // 删除的表
            List<DapDataModelTableDto> deleteTable = (List<DapDataModelTableDto>) map.get("deleteTable");
            // 最新变更的表
            List<DapDataModelTableDto> nowList = (List<DapDataModelTableDto>) map.get("nowList");
            // 旧版本表
            List<DapDataModelTableDto> list = (List<DapDataModelTableDto>) map.get("oldList");
            // 删除表
            if(!list.isEmpty()){
                deleteTable.addAll(list);
                addTable.addAll(nowList);
            }
            if(!deleteTable.isEmpty()){
                // 表id
                List<String> deleteIds = deleteTable.stream()
                        .map(DapDataModelTableDto::getTableId)
                        .collect(Collectors.toList());
                // 字段id
                List<String> fieldIds = new ArrayList<>();
                // 索引id
                List<String> keyIds = new ArrayList<>();
                deleteTable.forEach(d -> {
                    List<String> collect = d.getFieldList().stream()
                            .map(DapDataModelTableField::getId)
                            .collect(Collectors.toList());
                    fieldIds.addAll(collect);
                    List<String> keys = d.getKeyList().stream()
                            .map(DapDataModelSubordinateKey::getId)
                            .collect(Collectors.toList());
                    if(!keys.isEmpty()){
                        keyIds.addAll(keys);
                    }
                });
                // 删除表
                dapDataModelTableService.getBaseMapper().deleteBatchIds(deleteIds);
                // 删除字段
                dapDataModelTableFieldService.getBaseMapper().deleteBatchIds(fieldIds);
                // 删除索引
                if(!keyIds.isEmpty()){
                    dapDataModelTableFieldService.getBaseMapper().deleteBatchIds(keyIds);
                }
            }
            // 重新生成发布数据
            if(!addTable.isEmpty()){
                insertVersionData(addTable, databaseType);
            }
        }
        // 将发布数据的状态置为已发布
        modelRecordMapper.updateApplicationStatus(modelId);
        // 将旧版本放入版本历史表中


    }

    /**
     * v1.2
     * @param modelId
     * @param databaseType
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public ResultBean<Boolean> publishModel(String modelId, DatabaseType databaseType) {

        // 查询模型下属表字段
        List<DapDataModelTableDto> tableList = dapDataModelTableService.getCopyTableInfo(modelId);
        // 模型为空不允许发布
        if (tableList.isEmpty()) {
            return ResultBean.error("模型中的内容为空，无法发布！");
        } else {
            // 模型下的表为空，不允许发布
            for (DapDataModelTableDto dto : tableList) {
                if (StringUtils.isBlank(dto.getFieldList().get(0).getId())) {
                    return ResultBean.error("模型中表" + dto.getEngName() + "的内容为空，无法发布！");
                }
            }
        }
        List<DapDataModelTableDto> newList = getKeyAndPartitionList(tableList, databaseType, modelId, true);
        // 查询模型中是否存在发布记录
        List<DapDataModelRecord> dapDataModelRecords = getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelRecord>()
                        .eq(DapDataModelRecord::getModelId, modelId).eq(DapDataModelRecord::getBmStatus, BmStatus.ON.getCode()));
        // 第一次进行发布申请
        if(dapDataModelRecords.isEmpty()){
            // 插入版本数据
            Boolean aBoolean = insertVersionData(tableList, databaseType);
            if (!aBoolean) {
                return ResultBean.error("插入版本数据失败");
            }
            // 插入审批记录数据
            DapDataModelRecord dapDataModelRecord = VersionContrast.entity(modelId);
            boolean save = this.save(dapDataModelRecord);
            // 第一次进行发布，最后变更状态应该是新增表, 插入变更记录
            List<DapDataModelOperateHistory> histories = new ArrayList<>();
            for (DapDataModelTableDto dto : newList) {
                DapDataModelOperateHistory history = VersionContrast.historyList(dto, 1, databaseType);
                history.setDataModelRecordId(dapDataModelRecord.getId());
                histories.add(history);
            }
            dapDataModelOperateHistoryService.saveBatch(histories);
            return ResultBean.ok(save);
        }else{
            // 存在发布记录 若全部为已发布数据true
            boolean collect = dapDataModelRecords.stream().noneMatch(d -> d.getPublicStatus() != 1);
            if(!collect){
                return ResultBean.error("当前模型存在草稿或待审批或不通过状态，不允许发布");
            }
            // 如果表的更新时间大于模型的审核通过时间，表示进行了修改
            LocalDateTime max = dapDataModelRecords.stream().filter(o -> o.getReleaseTime() != null)
                    .map(DapDataModelRecord::getReleaseTime).distinct().max((e1, e2) -> e1.compareTo(e2)).get();
            // 存在已发布数据，但是不存在审核数据插入数据
            DapDataModelRecord dapDataModelRecord = VersionContrast.entity(modelId);
            // 查询上版本的数据
            List<DapDataModelTableDto> tables = dapDataModelTableService.getCopyTableList(modelId, 2);
            List<DapDataModelTableDto> oldList = getKeyAndPartitionList(tables, databaseType, modelId, false);
            ResultBean<List<DapDataModelOperateHistory>> result = VersionContrast.modelContrast(newList, databaseType, oldList, max, false);
            if (result.getCode() == 200) {
                boolean save = this.save(dapDataModelRecord);
                dapDataModelOperateHistoryService.updateRecordId(dapDataModelRecord.getId(), modelId);
                return ResultBean.ok(save);
            } else {
                return ResultBean.error(result.getMsg());
            }
        }
    }


    /**
     * 生成版本草稿数据
     * @param list
     * @return
     */
    public Boolean insertVersionData(List<DapDataModelTableDto> list, DatabaseType databaseType){

        // 生成表版本数据
        List<DapDataModelTable> tables = new ArrayList<>();
        List<DapDataModelTableField> fields = new ArrayList<>();
        List<DapDataModelSubordinateKey> keys = new ArrayList<>();
        List<DapDataModelTablePartition> partitions = new ArrayList<>();
        for(DapDataModelTableDto dto : list){
            // 表版本数据
            DapDataModelTable table = new DapDataModelTable();
            BeanUtils.copyProperties(dto, table);
            table.setPId(dto.getTableId());
            tables.add(table);
        }
        // 保存表版本数据
        dapDataModelTableService.saveBatch(tables);
        for(DapDataModelTable table : tables){
            for (DapDataModelTableDto dto : list){
                if(dto.getTableId().equals(table.getPId())){
                    // 新版本字段
                    List<DapDataModelTableField> collect = dto.getFieldList().stream()
                            .peek(f -> { f.setModelTableId(table.getId());f.setPId(f.getId());f.setId("");})
                            .collect(Collectors.toList());
                    fields.addAll(collect);
                    // 新版本分区
                    if(!dto.getPartitionList().isEmpty()){
                        List<DapDataModelTablePartition> partitionList = dto.getPartitionList().stream()
                                .peek(p -> {p.setModelTableId(table.getId());p.setPId(p.getId());p.setId("");})
                                .collect(Collectors.toList());
                        partitions.addAll(partitionList);
                    }
                    // 新版本索引
                    if(!dto.getKeyList().isEmpty()){
                        List<DapDataModelSubordinateKey> keyList = dto.getKeyList().stream()
                                .peek(k -> {k.setModelTableId(table.getId());k.setPId(k.getId());k.setId("");})
                                .collect(Collectors.toList());
                        keys.addAll(keyList);
                    }
                }
            }
        }
        // 保存字段版本数据
        boolean b = dapDataModelTableFieldService.saveBatch(fields);
        // 为hive数据源
        if(databaseType.equals(DatabaseType.HIVE)){
            // 保存分区版本信息
            return partitions.isEmpty() ? b : dapDataModelTablePartitionService.saveBatch(partitions);
        }else {
            // 为pg和mysql数据源
            if(!keys.isEmpty()){
                keys.forEach(key -> {
                    List<String> strings = Arrays.asList(key.getFieldIds().split(","));
                    String collect = fields.stream().filter(f -> strings.contains(f.getPId()))
                            .map(DapDataModelTableField::getId)
                            .collect(Collectors.joining(","));
                    key.setFieldIds(collect);
                });
                return dapDataModelSubordinateKeyService.saveBatch(keys);
            }
            return b;
        }
    }

    /**
     * 查询索引或者分区
     * @param list
     * @param databaseType
     * @return
     */
    public List<DapDataModelTableDto> getKeyAndPartitionList(List<DapDataModelTableDto> list,
                                                 DatabaseType databaseType, String modelId, boolean isNew){
        // hive数据库
        if(databaseType.equals(DatabaseType.HIVE)){
            // 查询该模型的最新的分区信息
            List<DapDataModelTablePartition> partitions = dapDataModelTablePartitionService.getBaseMapper()
                    .selectList(new LambdaQueryWrapper<DapDataModelTablePartition>()
                            .eq(DapDataModelTablePartition::getDataModelId, modelId)
                            .eq(DapDataModelTablePartition::getBmStatus, BmStatus.ON)
                            .eq(isNew ? DapDataModelTablePartition::getPId : DapDataModelTablePartition::getDataNewest, "0"));
            // 进行筛选
            list.forEach(t -> {
                List<DapDataModelTablePartition> collect = partitions.stream()
                        .filter(p -> p.getModelTableId().equals(t.getTableId()))
                        .collect(Collectors.toList());
                t.setPartitionList(collect);
            });
        }else{
            // 查询该模型下的索引信息
            List<DapDataModelSubordinateKey> dapDataModelSubordinateKeys = dapDataModelSubordinateKeyService
                    .getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelSubordinateKey>()
                            .eq(DapDataModelSubordinateKey::getDataModelId, modelId)
                            .eq(isNew ? DapDataModelSubordinateKey::getPId : DapDataModelSubordinateKey::getDataNewest, "0")
                            .eq(DapDataModelSubordinateKey::getBmStatus, BmStatus.ON));
            // 进行筛选
            list.forEach(t -> {
                List<DapDataModelSubordinateKey> collect = dapDataModelSubordinateKeys.stream().
                        filter(d -> d.getModelTableId().equals(t.getTableId()))
                        .collect(Collectors.toList());
                t.setKeyList(collect);
            });
        }
        return list;
    }
}
